<?php

namespace Php\Queue\work;

use Php\Queue\Query;
use Php\Queue\RedisDriver;
use Swoole\Process;
use function Swoole\Coroutine\run;

class Swoole extends QueueMaster
{
	public $query;
	
	/**
	 * 子进程数量
	 *
	 * @var int
	 */
	public $num;
	
	/**
	 * 记录子进程pid
	 *
	 * @var array
	 */
	public $workers = [];
	
	public $execute = 100000;
	
	public function __construct(Query $query)
	{
		$this->query = $query;
		$this->query->isSoole = false;
		$this->num = swoole_cpu_num();
//		$this->num = 20;
	}
	
	public function run()
	{
		$this->createChildProcess();
		$this->monitor();
	}
	
	private function createChildProcess()
	{
		while (count($this->workers) < $this->num) {
			$process = new Process(function (Process $proc) {
				echo '子进程' . getmypid() . '创建', PHP_EOL;
				run(function () use ($proc) {
					$this->workers($proc);
				});
				
			}, false, 2, false);
			$process->start();
			$this->workers[$process->pid] = $process->pid;
		}
	}
	
	private function workers(Process $proc)
	{
		$num = 0;
		$goCount = 0;
		$redis = RedisDriver::connect($this->query->config, 1);
		$boolean = true;
		while ($boolean) {
			// 将达到指定时间的消息加入到队列.
			$this->query->migrateExpiredJobs($redis, 100);
			// 当队列中没有消息时,将没有标记成功的消息恢复到队列中
			$this->query->recoverMessage($redis, 100);
			if ($num < $this->execute) {
				$result = $this->query->pop($redis);
				if ($result) {
					$goCount++;
					go(function () use ($result, &$goCount) {
						$class = $result['job'];
						unset($result['job']);
						$class = new $class;
						$class->job($this->query, $result);
						--$goCount;
					});
					$num++;
				} else {
					sleep(2);
				}
			} else {
				if ($goCount == 0) {
					$boolean = false;
				}
			}
		}
		if ($this->query->isSoole) {
			RedisDriver::giveBackRedis($redis);
		}
		$proc->exit(0);
	}
	
	private function monitor()
	{
		while ($ret = Process::wait()) {
			unset($this->workers[$ret['pid']]);
			$this->createChildProcess();
		}
	}
}